package com.atguigu.stream.veceivers

import java.io.{BufferedReader, InputStreamReader}
import java.net.Socket

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.receiver.Receiver

/**
 * description ：自定义数据源
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/1/13 18:07
 * modified By ：
 * version:    : 1.0
 */
class MyReceiver(host: String, port: Int) extends Receiver[String](StorageLevel.MEMORY_ONLY) {

  var socket: Socket = _
  var reader: BufferedReader = _

  // StreamingContex.start 之后调用
  /*
   接收器启动的时候调用该方法. This function must initialize all resources (threads, buffers, etc.) necessary for receiving data.
   这个函数内部必须初始化一些读取数据必须的资源
   该方法不能阻塞, 所以 读取数据要在一个新的线程中进行.
    */
  // TODO childThread函数中的子线程不能被阻塞，所以try catch要放到里面
  //  ！！！如果childThread中没有try catch,而是在外面定义了try catch 那么子线程发现异常，就会被阻塞
  override def onStart(): Unit = {
    childThread {
      try {
        socket = new Socket(host, port)
        reader = new BufferedReader(new InputStreamReader(socket.getInputStream, "UTF-8"))
        var line = reader.readLine()
        while (socket.isConnected && null != line) {
          store(line)
          line = reader.readLine()
        }
      } catch {
        case e => println(e.getMessage)
      }
      finally {
        restart("restart socket ......")
      }
    }
  }
  def childThread(f: => Unit) {
    new Thread("Socket Receiver") {
      override def run(): Unit = f
    }.start()
  }
  // 停止的时候调用，释放资源
  override def onStop(): Unit
  = {
    if (null != socket) socket.close()
    if (null != reader) reader.close()
  }
}


object MyReceiver {
  def apply(host: String, port: Int) = new MyReceiver(host, port)
}
